package com.zhao.demo.unbound.demo04_transformation.sample07_self.a_common

import com.zhao.demo.Raytek
import org.apache.flink.api.common.functions.FilterFunction
import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment

/**
 * Description: 用户自定义函数之子类方式<br/>
 * Copyright (c) ，2020 ， 赵 <br/>
 * This program is protected by copyright laws. <br/>
 * Date： 2020/12/9 10:05
 * 需求:筛选出当天所有途经红外测温仪编号为raytek_2的旅客信息
 *
 * @author 柒柒
 * @version : 1.0
 */

object SelfDefineFunctionDemo {
  def main(args: Array[String]): Unit = {

    //1.前提 拦截非法的参数
    if (args == null || args.length != 6){
      sys.error(
        """
          |请传入参数! --hostname <主机名> --port <端口号> --id <红外测温仪的id>
          |""".stripMargin)
      sys.exit(-1)
    }

    //2.获得参数值
    val tool: ParameterTool = ParameterTool.fromArgs(args)
    val hostname = tool.get("hostname")
    val port = tool.getInt("port")
    val id = tool.get("id")

    //3.环境
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    //4.实时读取流数据，计算，并显示结果
    import org.apache.flink.api.scala._

    env.socketTextStream(hostname,port)
      .map(perInfo =>{
        val arr: Array[String] = perInfo.split(",")
        val id = arr(0).trim
        val temperature = arr(1).trim.toDouble
        val name = arr(2).trim
        val timestamp = arr(3).trim.toLong
        val location = arr(4).trim
        Raytek(id, temperature, name, timestamp, location)
      }).filter(new MyFilter(id))
      .print("使用自定义函数子类,实现的效果->")
  }

  /**
   * 自定义过滤器的子类
   */
  class MyFilter(id: String) extends FilterFunction[Raytek]{
    /**
     * 在下述方法中书写过滤业务逻辑
     * 挑选当天所有途径红外测温仪编号为raytek_2的旅客的信息
     * @param value
     * @return
     */
    override def filter(value: Raytek): Boolean = {
      id.equals(value.id)
    }
  }
}

